package com.zslc.securitymonitoring.websocket;

import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.zslc.securitymonitoring.config.GetHttpSessionConfigurator;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 今日事件推送
 * @author Gaoyc
 * @date 2024/06/16 14:08
 */
@Component
@ServerEndpoint(value = "/realtime/display", configurator = GetHttpSessionConfigurator.class)
@Slf4j
public class WebSocketServer {

    //用来记录当前在线连接数，线程安全。
    private static final AtomicInteger onlineCount = new AtomicInteger(0);

    //线程安全map，存放每个客户端的MyWebSocket对象。
    private static final ConcurrentHashMap<String, WebSocketServer> webSocketMap = new ConcurrentHashMap<>();
    private static final ConcurrentHashMap<String, String> sessionToBusIdMap = new ConcurrentHashMap<>();

    //与某个客户端的连接会话
    private Session session;
    public Session getSession() {
        return session;
    }
    public void setSession(Session session) {
        this.session = session;
    }

    public static synchronized int getOnlineCount() {
        return onlineCount.get();
    }

    public static synchronized void addOnlineCount() {
        WebSocketServer.onlineCount.incrementAndGet();
    }

    public static synchronized void subOnlineCount() {
        WebSocketServer.onlineCount.decrementAndGet();
    }

    /**
     * 连接建立成功调用的方法
     */
    @OnOpen
    public void onOpen(Session session, EndpointConfig config) {
        this.session = session;

        webSocketMap.put(session.getId(), this);
        addOnlineCount();           //在线数加1
        log.info("websocket有新连接加入！当前在线连结数为" + getOnlineCount());
        try {
//            String busId = (String) config.getUserProperties().get("busId");
            String busId = "admin";
            log.info("websocket添加通道，业务id：{}", busId);
            sessionToBusIdMap.put(session.getId(), busId);

        } catch (Exception e) {
            log.error("onOpen异常",e);
        }
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose(Session session) {
        String busId = sessionToBusIdMap.get(session.getId());

        //从webSocketMap中删除
        webSocketMap.remove(session.getId());
        subOnlineCount();           //在线数减1
        log.info("websocket任务:【{}】退出！当前在线连接数为{}",busId, getOnlineCount());
        sessionToBusIdMap.remove(session.getId());
    }

    /**
     * 收到客户端消息后调用的方法
     *
     * @param message 客户端发送过来的消息
     */
    @OnMessage
    public void onMessage(String message, Session session) {
        log.info("========onMessage:{}",message);
        if (StringUtils.isNotEmpty(message)) {
            String busId = sessionToBusIdMap.get(session.getId());
            log.info("websocket接收到【{}】消息，返回【{}】。",busId, message);
        }

    }

    /**
     * 发生错误时调用
     *
     * @OnError
     **/
    @OnError
    public void onError(Session session, Throwable error) {
        String busId = sessionToBusIdMap.get(session.getId());
        log.error("websocket发生错误：{}，busId:【{}】", error.getMessage(), busId);
        error.printStackTrace();
    }


    /**
     * 向客户端发送消息
     *
     * @param message 消息内容
     */
    public void sendMessage(Object message) throws IOException {
        String jsonMsg = JSONObject.toJSONString(message);
        this.session.getAsyncRemote().sendText(jsonMsg);
    }

    public void sendMessage(String busId, Object message) {
        try {
            log.info("开始websocket推送，通道id：{}", busId);
            log.info("开始websocket推送，消息：{}", message != null ? JSON.toJSONString(message) : null);
            List<String> sessionIdList = getSessionIdByBusId(busId);
            log.info("websocket通道集合：{}", CollectionUtil.isNotEmpty(sessionIdList) ? sessionIdList.toString() : null);
            sessionIdList.forEach(s -> {
                WebSocketServer webSocketServer = webSocketMap.get(s);
                if (!Objects.isNull(webSocketServer)) {
                    try {
                        webSocketServer.sendMessage(message);
                    } catch (IOException e) {
                        throw new RuntimeException(e);
                    }
                    log.info("websocket推送成功，通道id：{}", busId);
                }
            });
        } catch (Exception e) {
            log.error("websocket推送失败，通道id：{}，错误：{}", busId, e);
        }
    }

    public List<String> getSessionIdByBusId(String busId) {
        List<String> sessionIds = new ArrayList<>();
        for (Map.Entry<String, String> sb : sessionToBusIdMap.entrySet()) {
            if (sb.getValue().equals(busId)) {
                sessionIds.add(sb.getKey());
            }
        }
        return sessionIds;
    }

    /**
     * 群发自定义消息
     */
    public void broadCastInfo(Object message) {
        try {
            for (WebSocketServer item : webSocketMap.values()) {
                item.sendMessage(message);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
